package io.polly.ldz.reactor.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * @Author ldz112@163.com
 * @Classname KafkaCustomizePartitioner
 * @Version V1.0
 * @Description 自定义分区
 * @Date 2024/9/29 21:28
 * Copyright (C) 2023 All Rights Reserved
 */
@Slf4j
public class KafkaCustomizePartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        //获取分区列表
        List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
        int partitionNum = 0;
        if(key == null){
            partitionNum = new Random().nextInt(partitionInfos.size());//随机分区
        } else {
            Long keyLong  = Long.parseLong(String.valueOf(key));
            partitionNum =  keyLong.intValue() % partitionInfos.size();
        }
        log.info("当前Key:" + key + "-----> 当前Value:" + value + "----->" + "当前存储分区:" + partitionNum);
        return partitionNum;

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
